package cn.sheep.dmp.report

import java.util.Properties

import cn.sheep.dmp.utils.Tools
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{SQLContext, SaveMode}

/**
  * 统计广告在地域维度上的指标分布情况
  * Sheep.Old @ 64341393
  * Created 2018/3/29
  */
object Report4AreaSQL {

    def main(args: Array[String]): Unit = {

        val sparkConf = new SparkConf().setAppName("地域维度指标分布")
          .setMaster("local[*]")
          .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

        val sc = new SparkContext(sparkConf)
        val sqlc = new SQLContext(sc)

        // 读parquet文件
        val dataFrame = sqlc.read.parquet(Tools.load.getString("parquet.path"))

        dataFrame.registerTempTable("logs")


        // 注册一个udf 多对多  udaf 多对一
        sqlc.udf.register("oxox", (b: Boolean) => if (b) 1 else 0)


        val result = sqlc.sql(
            """
              |select
              |provincename, cityname,
              |sum(oxox(requestmode=1 and processnode>=1)) rawReq,
              |sum(case when requestmode=1 and processnode>=2 then 1 else 0 end) effReq,
              |sum(case when requestmode=1 and processnode=3 then 1 else 0 end) adReq,
              |sum(case when iseffective=1 and isbilling=1 and isbid=1 and adorderid!=0 then 1 else 0 end) rtbReq,
              |sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1 else 0 end) succReq,
              |sum(case when requestmode=2 and iseffective=1 then 1 else 0 end) adShow,
              |sum(case when requestmode=3 and iseffective=1 then 1 else 0 end) adClick,
              |sum(case when iseffective=1 and isbilling=1 and iswin=1 then winprice/1000 else 0 end) xf,
              |sum(case when iseffective=1 and isbilling=1 and iswin=1 then adpayment/1000 else 0 end) cb
              |from logs
              |group by provincename, cityname
            """.stripMargin)

        // 将计算结果写入到数据库中
        val props = new Properties()
        props.setProperty("user", Tools.load.getString("db.user"))
        props.setProperty("password", Tools.load.getString("db.password"))

        result.write.mode(SaveMode.Append).jdbc(
            Tools.load.getString("db.url"),
            "report_area_kpi1",
            props
        )


        sc.stop()
    }



}
